perf(core): run visual participants (actor, scene) in parallel within a shared decode stream - #129
Conversation
… shared decode stream Refactors _consume_visual_stream so the decode loop runs on a thread and each visual participant (actor via OpenCV, scene via torch) gets its own worker thread consuming from a per-participant queue. This lets actor and scene overlap on CPU since both libraries release the GIL during computation. - Decode thread pushes RGB batches to bounded queues (maxsize=4) for backpressure - Each participant worker filters its own samples and calls process() independently - Exceptions from any thread propagate via a shared error list - Cancellation checked in both decode and participant workers with a 0.2s polling timeout on the queue get to stay responsive - The single-decode property (one pass through iter_frame_batches) is preserved; actor stays in-order (single FIFO consumer) No changes to runner.py, IndexConfig, ResourceScheduler, or any capability's indexing logic. All 5 new threading tests pass.
- Fix a critical shutdown deadlock when a participant queue is full: the sentinel drops, so workers now also exit once decoding is done and their queue is empty. - Stop workers promptly after a sibling records an error instead of processing remaining queued batches. - Normalize duplicate modality names in index_visuals before building participant/reporting structures. - Reorder the cancel timer callback so cancellation happens before unblocking the stream. - Prefer addClassCleanup over tearDownClass and drop an unused rebinding. Internal-only.
There was a problem hiding this comment.
Thanks for taking this on. The shared stream implementation and queue/shutdown tests are a useful foundation.
I think the PR description needs one clarification: the existing implementation already decoded a single shared frame stream for all visual participants, so this change parallelizes participant processing rather than removing multiple decode passes.
Before merging, could we also add before/after latency and memory measurements for both scene alone and scene,actor? The discussion on issue #96 placed the benchmark first, and the scene-only case would help show the overhead of the new thread and queues when only one participant is selected.
| if errors: | ||
| return | ||
| cancellation.raise_if_cancelled() | ||
| report_progress( |
There was a problem hiding this comment.
One production path may need adjustment here. These worker threads can call the progress and cancellation callbacks outside the thread that owns the DBOS workflow context. The production progress callback uses DBOS.set_event(), while DBOS keeps the active workflow state in a ContextVar that raw threading.Threads do not inherit. This means the first progress update may fail the indexing job. The current tests do not exercise that path because they use progress=None and a plain cancellation token. Could we keep these callbacks on the owning thread, or carry the required context into the workers, and add a production-style callback test?
2 participants
Summary
Runs the
actorandscenevisual participants in parallel over a single shared decode stream: one decode pass, per-participant bounded queues, and sentinel-based shutdown. Actor remains in-order; backpressure stalls decode when a participant falls behind. Also fixes a deadlock where a participant failing mid-stream would hang the pipeline — the participant error now propagates instead.Internal-only; no public API or storage changes.
Motivation
Refs
grayhatdevelopers#96. Video decoding is the common cost shared by the visual participants (scene,actor). Previously each participant ran its own pass over the decoded frames, so N participants meant N decode passes. This change decodes once and feeds both participants concurrently, cutting redundant decode work and letting the slower participant overlap with the faster one.What changed
src/vidxp/capabilities/visual.py— parallel participant scheduling over a shared decode stream:sceneandactorconsumers run concurrently in their own workers;actorremains in-order whilescenemay lag;tests/test_visual_threading.py— 7 unit tests covering concurrent processing, cancellation, decode failures, participant failures, ordering, and backpressure/shutdown.Design details
Validation
uv run --no-sync python -m pytest -q tests/test_visual_threading.py— 7 passed.uv run --no-sync python -m pytest -q tests/test_runner.py tests/test_manifest.py tests/test_generation_manifest.py— passed (reported 27 in the fork PR).uv run --no-sync ruff check src/vidxp/capabilities/visual.py tests/test_visual_threading.py— clean.main(c17a884); the working tree is clean and the visual-area tests pass on the rebased base.Notes
Internal-only performance/reliability change. No public API, storage schema, CLI, or MCP surface changes.